package com.liu.audio.controller;

import com.liu.audio.domain.CosyVoice;
import com.alibaba.dashscope.audio.tts.SpeechSynthesisResult;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesisAudioFormat;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesisParam;
import com.alibaba.dashscope.audio.ttsv2.SpeechSynthesizer;
import com.alibaba.dashscope.common.ResultCallback;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.ThreadUtils;
import org.apache.commons.lang3.StringUtils;
import org.jetbrains.annotations.NotNull;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.reactive.function.client.WebClient;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Scheduler;
import reactor.core.scheduler.Schedulers;

import java.nio.ByteBuffer;
import java.time.Duration;
import java.util.HashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

@Slf4j
@RestController
@RequestMapping("/sse/mitter")
public class SseMitterController {

    @GetMapping(value = "/stream", produces = "text/event-stream")
    public SseEmitter stream() {
        // 设置默认超时时间  0L 表示无限
        // 注意：这里的单位是  ms
        SseEmitter sseEmitter = new SseEmitter(30000L);

        // 最好不要阻塞主线程
        Executors.newSingleThreadExecutor().execute(() -> {
            try {
                for (int i = 0; i < 10; i++) {
                    sseEmitter.send("这只是一个流式输出案例：" + i);
                    TimeUnit.SECONDS.sleep(1);
                }
                // 通知客户端消息发送完毕
                sseEmitter.complete();
            } catch (Exception e) {
                e.printStackTrace();
                sseEmitter.completeWithError(e);
            }
        });

        return sseEmitter;
    }

    @PostMapping(value = "/stream-sse-sink-tts", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<CosyVoice>> audioTTS(@RequestBody String message) {
        //@wjw_comment: 必须是unicast().onBackpressureError(),否则服务的收不到断开事件
        // unicast() 提供只能一个订阅者的单播
        // onBackpressureError() 拒绝第一个订阅者之后的其它订阅者
        Sinks.Many<ServerSentEvent<CosyVoice>> sink = Sinks.many().unicast().onBackpressureError();

        Flux<ServerSentEvent<CosyVoice>> flux = sink.asFlux();

        // Prepare the speech synthesis task
        SpeechSynthesisParam param = SpeechSynthesisParam.builder()
                // 若没有将API Key配置到环境变量中，需将下面这行代码注释放开，并将apiKey替换为自己的API Key
                .apiKey(apikey)
                .model(model)
                .voice(voice)
                .format(SpeechSynthesisAudioFormat.PCM_22050HZ_MONO_16BIT)
                .build();
        SpeechSynthesizer synthesizer = new SpeechSynthesizer(param, new ReactCallback(sink));

        getFlux(message).subscribe(i -> {
            log.info(i);
            if (!"[DONE]".equals(i)) {
                JSONObject jsonObject = JSON.parseObject(i, JSONObject.class);
                JSONObject res = jsonObject.getJSONObject("message");
                String content = res.getString("content");
                if (StringUtils.isNotBlank(content)) {

                    // 调用语音TTS
                    synthesizer.streamingCall(content);

                    // 对话内容返回
                    CosyVoice cosyVoice = new CosyVoice(1, content, null);
                    ServerSentEvent<CosyVoice> serverSentEvent = ServerSentEvent.<CosyVoice>builder()
                            //  .id(String.valueOf(i))
                            //  .event("periodic-event")
                            .data(cosyVoice)
                            .build();

                    sendMessage(sink, serverSentEvent);
                }
            }else {
                synthesizer.streamingComplete();
            }
        });


        return flux;
    }

    private static void sendMessage(Sinks.Many<ServerSentEvent<CosyVoice>> sink, ServerSentEvent<CosyVoice> serverSentEvent) {
//        sendMessage(sink, serverSentEvent,0);
        if (sink.tryEmitNext(serverSentEvent).isFailure()) {
            try {
                Thread.sleep(50);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

            if (sink.tryEmitNext(serverSentEvent).isFailure()) {
                log.error("sink.tryEmitNext isFailure :{} ", serverSentEvent);
            }
        }
    }

    private static void sendMessage(Sinks.Many<ServerSentEvent<CosyVoice>> sink, ServerSentEvent<CosyVoice> serverSentEvent, int retry) {
        if (sink.tryEmitNext(serverSentEvent).isFailure()) {
            if (retry > 3) {
                log.error("sink.tryEmitNext isFailure :{} ", serverSentEvent);
            } else {
                retry++;
                sendMessage(sink, serverSentEvent, retry);
            }
        }
    }

    private static @NotNull Flux<String> getFlux(String message) {
        String url = "http://localhost:11434/api/chat";
        String model = "deepseek-r1:8b";
        // 构建请求体
        HashMap<String, Object> params = new HashMap<>();
        params.put("model", model);
        params.put("stream", true);
        JSONArray messages = new JSONArray();
        params.put("messages", messages);

        JSONObject obj = new JSONObject();
        obj.put("role", "user");
        obj.put("content", message);
        messages.add(obj);

        WebClient webClient = WebClient.create();
        // 输出响应结果
        return webClient.post()
                .uri(url)
                .header(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
//                .header(HttpHeaders.AUTHORIZATION, "Bearer "+ API_KEY) // 添加认证头部
                .bodyValue(params)
                .retrieve()
                .bodyToFlux(String.class);
    }


    // 使用 Sinks.Many<ServerSentEvent<String>> 对应非反应式的SseEmitter
    @GetMapping(value = "/stream-sse-sink", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public Flux<ServerSentEvent<CosyVoice>> streamSseMvc() {
        //@wjw_comment: 必须是unicast().onBackpressureError(),否则服务的收不到断开事件
        // unicast() 提供只能一个订阅者的单播
        // onBackpressureError() 拒绝第一个订阅者之后的其它订阅者
        Sinks.Many<ServerSentEvent<CosyVoice>> sink = Sinks.many().unicast().onBackpressureError();

        Flux<ServerSentEvent<CosyVoice>> flux = sink.asFlux();


        Scheduler single = Schedulers.boundedElastic();
        single.schedule(() -> {
            // Prepare the speech synthesis task
            SpeechSynthesisParam param = SpeechSynthesisParam.builder()
                    // 若没有将API Key配置到环境变量中，需将下面这行代码注释放开，并将apiKey替换为自己的API Key
                    .apiKey(apikey)
                    .model(model)
                    .voice(voice)
                    .format(SpeechSynthesisAudioFormat.PCM_22050HZ_MONO_16BIT)
                    .build();
            SpeechSynthesizer synthesizer = new SpeechSynthesizer(param, new ReactCallback(sink));


            String str = "今天天气怎么样？";
            for (int i = 0; i < str.length(); i++) {
                String text = String.valueOf(str.charAt(i));
                synthesizer.streamingCall(text);

                try {
                    ThreadUtils.sleep(Duration.ofMillis(100));
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }

            synthesizer.streamingComplete();
        });

        return flux;
    }


    private static String apikey = "--------";
    private static String model = "cosyvoice-v1";
    private static String voice = "longxiaochun";

    // Create a subclass inheriting from ResultCallback<SpeechSynthesisResult>
    // to implement the callback interface
    // 创建一个继承自ResultCallback<SpeechSynthesisResult>的子类来实现回调接口
    static class ReactCallback extends ResultCallback<SpeechSynthesisResult> {
        Sinks.Many<ServerSentEvent<CosyVoice>> sink;

        public ReactCallback(Sinks.Many<ServerSentEvent<CosyVoice>> sink) {
            this.sink = sink;
        }

        // Callback when the service side returns the streaming synthesis result
        // 当服务侧返回流式合成结果后回调
        @Override
        public void onEvent(SpeechSynthesisResult result) {
            // Get the binary data of the streaming result via getAudio
            // 通过getAudio获取流式结果二进制数据
            if (result.getAudioFrame() != null) {
                ByteBuffer audioFrame = result.getAudioFrame();
                byte[] data = audioFrame.array();
                CosyVoice cosyVoice = new CosyVoice(1, "", data);

                ServerSentEvent<CosyVoice> serverSentEvent = ServerSentEvent.<CosyVoice>builder()
                        //  .id(String.valueOf(i))
                        //  .event("periodic-event")
                        .data(cosyVoice)
                        .build();

                log.info("stream-sse-sink: " + serverSentEvent);
                sendMessage(sink, serverSentEvent);

            }
        }

        // Callback when the service side completes the synthesis
        // 当服务侧完成合成后回调
        @Override
        public void onComplete() {
            // Notify the playback thread to end
            sink.tryEmitComplete();
        }

        // Callback when an error occurs
        // 当出现错误时回调
        @Override
        public void onError(Exception e) {
            // Tell the playback thread to end
            e.printStackTrace();
            sink.tryEmitComplete();
        }
    }

}
